package org.example.slave.channel;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.util.internal.OutOfDirectMemoryError;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;
import lombok.Data;
import lombok.SneakyThrows;

import java.net.SocketAddress;
import java.util.Objects;

/**
 * @author dingrui
 * @since 2022/2/16
 * @description 将netty客户端的channel实例持有在该类中
 */
@Data
public class NettyChannel implements MyChannel {

    private static final int DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024; // 缓存初始 1M
    private static final int DEFAULT_MAX_BUFFER_SIZE = 16 * DEFAULT_INIT_BUFFER_SIZE; // 最大缓存 16M

    // netty的channel
    private Channel ch;

    private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存
    private int maxDirectBuffer = this.cache.maxCapacity();

    private final Object lock = new Object();
    private static final int WAIT_PERIOD = 10; // ms

    private NettyChannel() {
    }

    public NettyChannel(Channel ch) {
        this.ch = ch;
    }

    @Override
    public void write(byte[]... buf) {
        if (Objects.isNull(this.ch) || !this.ch.isWritable()) throw new RuntimeException("write failed");
        this.ch.writeAndFlush(Unpooled.copiedBuffer(buf));
    }

    @Override
    public byte[] read(int readSize) {
        return this.read(readSize, 0L);
    }

    @Override
    public byte[] read(int readSize, long timeout) {
        int accumulatedWaitTime = 0;
        // 若读取内容较长 则自动扩充超时时间 以初始缓存大小为基准计算倍数
        if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE) {
            timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1);
        }
        do {
            if (readSize <= this.cache.readableBytes()) { // 不需要等待缓存中数据
                byte[] back = new byte[readSize];
                synchronized (lock) {
                    cache.readBytes(back);
                }
                return back;
            } else {
                if (Objects.isNull(this.ch))
                    throw new RuntimeException("channel is closed"); // 因为要等缓存后续的持续数据 如果channel关闭了 即使将现有的缓存读走 最终数据也不完整
                if (timeout > 0) { // 持续等待缓存新写的数据
                    accumulatedWaitTime += WAIT_PERIOD;
                    if (accumulatedWaitTime > timeout) throw new RuntimeException("read timeout");
                }
                synchronized (this) {
                    try {
                        wait(WAIT_PERIOD);
                    } catch (InterruptedException e) {
                        throw new RuntimeException("channel is interrupted");
                    }
                }
            }
        } while (true);
    }

    @Override
    public void read(byte[] data, int off, int len, long timeout) {
        // TODO: 2022/2/16
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isConnected() {
        return Objects.nonNull(this.ch);
    }

    @Override
    public SocketAddress getRemoteSocketAddress() {
        return Objects.isNull(this.ch) ? null : this.ch.remoteAddress();
    }

    @Override
    public SocketAddress getLocalSocketAddress() {
        return Objects.isNull(this.ch) ? null : this.ch.localAddress();
    }

    @Override
    public void close() {
        if (Objects.nonNull(this.ch)) this.ch.close();
        this.ch = null;
        synchronized (this.lock) {
            this.cache.discardReadBytes(); // 回收已经占用内存
            this.cache.release(); // 释放整内存
            this.cache = null;
        }
    }

    // 接收服务端的响应 缓存在内存中
    @SneakyThrows
    @Override
    public void writeCache(ByteBuf byteBuf) {
        synchronized (this.lock) {
            while (true) {
                if (Objects.isNull(this.cache)) throw new RuntimeException("channel is closed");
                if (!byteBuf.isReadable()) break;
                // 默认缓存大小不够用时需自动清理或扩充，否则将因缓存空间不足而造成I/O超时假象
                int length = byteBuf.readableBytes();
                int deltaSize = length - this.cache.writableBytes();
                if (deltaSize > 0) { // 缓存中现有内存大小不够 需要扩容
                    if (this.cache.readerIndex() >= deltaSize) { // 优先清理缓存中数据
                        // 回收已读空间，重置读写指针
                        this.cache.discardReadBytes();
                        // 恢复自动扩充的过大缓存到默认初始缓存大小 释放空间
                        int oldCapacity = this.cache.capacity();
                        if (oldCapacity > DEFAULT_MAX_BUFFER_SIZE) { // 尝试收缩
                            int newCapacity = this.cache.writerIndex();
                            newCapacity = ((newCapacity - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            int quarter = (newCapacity >> 2); // 至少留空四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            newCapacity += quarter; // 留空四分之一
                            if (newCapacity < (oldCapacity >> 1)) { // 至少收缩二分之一
                                try {
                                    this.cache.capacity(newCapacity);
                                } catch (OutOfMemoryError ignore) {
                                    this.maxDirectBuffer = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                }
                            }
                        }
                    } else { // 尝试扩容
                        int oldCapacity = this.cache.capacity();
                        if (oldCapacity < this.maxDirectBuffer) {
                            int quarter = (oldCapacity >> 2); // 至少扩容四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            deltaSize = ((deltaSize - 1) / quarter + 1) * quarter; // 对齐
                            int newCapacity = oldCapacity + deltaSize;
                            if (newCapacity > maxDirectBuffer) newCapacity = maxDirectBuffer;
                            try {
                                this.cache.capacity(newCapacity);
                            } catch (OutOfDirectMemoryError e) {
                                long maxDirectMemory = SystemPropertyUtil.getLong("io.netty.maxDirectMemory", -1);
                                if (maxDirectMemory < 0) maxDirectMemory = PlatformDependent.maxDirectMemory();
                                if (this.maxDirectBuffer > maxDirectMemory) {
                                    this.maxDirectBuffer = (int) maxDirectMemory;
                                    newCapacity = this.maxDirectBuffer;
                                    try {
                                        this.cache.capacity(newCapacity);
                                    } catch (OutOfMemoryError ignore) {
                                        this.maxDirectBuffer = oldCapacity; // 未来不再超过当前容量
                                    }
                                } else {
                                    this.maxDirectBuffer = oldCapacity; // 未来不再超过当前容量
                                }
                            } catch (OutOfMemoryError e) {
                                // ignore
                                this.maxDirectBuffer = oldCapacity; // 未来不再超过当前容量
                            }
                        }
                    }
                    deltaSize = length - this.cache.writableBytes();
                }

                if (deltaSize != length) {
                    // deltaSize <= 0 可全部写入，deltaSize > 0 只能部分写入
                    if (deltaSize <= 0) {
                        this.cache.writeBytes(byteBuf, length);
                        break;
                    } else
                        this.cache.writeBytes(byteBuf, length - deltaSize);
                }
                this.lock.wait(WAIT_PERIOD);
                // 回收已读空间 重置读写指针
                this.cache.discardReadBytes();
            }
        }
    }
}
